package com.wxq.ware.constant;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author Jim
 * @date 8:02 下午 2022/4/20
 * @description
 * @email:huwmz04153277@126.com
 */
@Component
public class Consumer {
    //配置topic和分区,可以配置多个
//topic为队列名称
//partitions表示值的的分区，这里指定了0和2分区
//partitionOffsets表示详细的指定分区，partition表示那个分区，initialOffset表示Offset的初始位置
    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
  //  @KafkaListener(topics = "test", groupId = "zhTestGroup")
    @KafkaListener(topicPartitions =
            { @TopicPartition(topic = "test",
                    partitions = { "0", "2" },
                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "4"))
            }
    )
    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.err.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }
}
